package com.timeriver.feature_select

import org.apache.spark.ml.feature.{ChiSqSelector, VectorAssembler}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
  * 对具有分类特征的标记数据进行操作。ChiSqSelector使用独立性的皮尔森卡方检验来决定选择哪些特征。
  * 它支持五种选择方法：更换默认值时使用setSelectorType来切换
  *   1. numTopFeatures，根据Chi-squared检验选择固定数量的顶级特征。这类似于产生具有最大预测能力的特征。
  *   2. percentile, 参数在0-1之间
  *   3. fpr, 选择所有pValue低于阈值的特征，从而控制选择的假阳性率
  *   4. fdr, 使用Benjamini-Hochberg程序来选择所有错误发现率低于阈值的特征
  *   5. fwe，选择所有p值低于阈值的特征。阈值以1/numFeatures为尺度，从而控制选择的家族式错误率。默认情况下，
  *           选择方法是numTopFeatures，默认的顶级特征数量设置为50。用户可以使用setSelectorType来选择选择方法。
  */
object ChiSqSelectorDemo {
  def main(args: Array[String]): Unit = {
    val session: SparkSession = SparkSession.builder()
      .appName("特征选择算子")
      .master("local[6]")
      .getOrCreate()

    val df: DataFrame = session.read
      .format("jdbc")
      .option("url", "jdbc:mysql://10.0.24.197:3306/ml_datasets")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("dbtable", "breast_cancer_wisconsin")
      .option("user", "root")
      .option("password", "123456")
      .load()

    /** 过滤缺失值 */
    val value: Dataset[Row] = df.filter(!_.anyNull)

    /** 获取特征列字段数组 */
    val inputCols: Array[String] = "clump_thickness,uniformity_of_cell_size,uniformity_of_cell_shape,marginal_adhesion,single_epithelial_cell_size,bare_nuclei,blan_chromatin,normal_nucleoli,mitoses".split(",")

    /** 构建特征列向量 */
    val data: DataFrame = new VectorAssembler()
      .setInputCols(inputCols)
      .setOutputCol("features")
      .transform(value)

    /** 特征选择 */
    val selector: ChiSqSelector = new ChiSqSelector()
//      .setNumTopFeatures(7)
      .setSelectorType("percentile")
      .setPercentile(0.3)
      .setFeaturesCol("features")
      .setLabelCol("class")
      .setOutputCol("selectedFeatures")

    val frame: DataFrame = selector.fit(data).transform(data)
    frame.show(false)
  }
}
